/*
 * Copyright (c) 2025 Li Auto Inc. and its affiliates
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include "RAWChannelResource.h"
#include <cstring>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <transport/ReceiverResource.h>
#include <deps/common/IPFinder.h>

#include <arpa/inet.h>

#define MAX_IP_HEADER_LENGTH 60

namespace vbs {
namespace transport {

using IPFinder = vbsutil::xmlparser::IPFinder;
using IPLocator = vbsutil::xmlparser::IPLocator;

RAWChannelResource::RAWChannelResource(uint32_t maxMsgSize, const Locator& locator, const std::string& sInterface,
                                       std::shared_ptr<ReceiverResource> receiver)
    : ChannelResource(maxMsgSize), message_receiver_(std::move(receiver)), socket_(0), interface_(sInterface) {
    std::stringstream ss;
    // 收包线程名格式为R + ip + port. 其中ip为16进制, 固定占8个字节. port为10进制
    ss << "R" << std::hex << std::setw(8) << std::setfill('0') << ntohl(inet_addr(sInterface.c_str()))
       << std::to_string(locator.port);
    std::string thread_name(ss.str());

    elogInfo(TRANSPORT,
             "RAWChannelResource  CREATE locator " << locator << "sInterface.c_str(): " << sInterface.c_str());

    int sock = socket(AF_INET, SOCK_RAW, IPPROTO_UDP);
    if (sock == -1) {
        elogWarning(RTPS_MSG_OUT, "Error socket: " << std::strerror(errno) << " (errno: " << errno << ")");
        return;
    }

    set_socket(sock);

    const uint32_t ip_addr = inet_addr(sInterface.c_str());
    struct sock_filter BPF_code[] = {
        // Load the 9th byte of the IP header (protocol)
        {0x30, 0, 0, 0x00000009},  // Load 1 byte from offset 9
        // // Check if it is UDP (17)
        {0x15, 0, 4, 0x00000011},  // Jump to end if not UDP
        // Load destination IP address (use hardcoded offset for IPv4)
        {0x28, 0, 0, 0x00000010},      // Load 4 bytes from offset 16
        {0x45, 0, 2, htonl(ip_addr)},  // Jump 2 if not equal to ip_addr
        // Load RTPS
        {0x28, 0, 0, 0x0000001c},         // Load 4 bytes from offset 28
        {0x45, 1, 0, htonl(0x52545053)},  // Jump 1 if equal to RTPS
        // Not matching
        {0x6, 0, 0, 0x00000000},  // Reject packet
        // Accept the packet
        {0x6, 0, 0, 0x00040000}  // Accept packet
    };

    struct sock_fprog filter;
    filter.len = sizeof(BPF_code) / sizeof(BPF_code[0]);
    filter.filter = BPF_code;

    if (setsockopt(sock, SOL_SOCKET, SO_ATTACH_FILTER, &filter, sizeof(filter)) == -1) {
        elogError(RTPS_MSG_OUT, RetCode_t::RETCODE_ERROR,
                  "Error setsockopt SO_ATTACH_FILTER: " << std::strerror(errno) << " (errno: " << errno << ")");
        close(sock);
        return;
    }

    thread(std::thread(&RAWChannelResource::perform_listen_operation, this, locator, thread_name));
}

RAWChannelResource::~RAWChannelResource() {
    message_receiver_ = nullptr;
    close(get_socket());
}

void RAWChannelResource::perform_listen_operation(Locator input_locator, const std::string& thread_name) {
    Locator remote_locator;
    Locator local_locator;
    logDebug(RTPS_MSG_IN, "perform_listen_operation  start for  " << input_locator);

    (void)thread_name;
    (void)pthread_setname_np(pthread_self(), thread_name.c_str());

    while (alive()) {
        // Blocking receive.
        unsigned char* buffer = message_buffer();
        uint32_t length = 0;
        if (!Receive(buffer, buffer_size(), length, remote_locator, local_locator)) {
            continue;
        }
        logDebug(RTPS_MSG_IN, "perform_listen_operation  receive msg " << length);
        // Processes the data through the CDR Message interface.
        if (message_receiver() != nullptr) {
            message_receiver()->OnDataReceived(buffer, length, local_locator, remote_locator);
        } else if (alive()) {
            logDebug(RTPS_MSG_IN, "Received Message, but no receiver attached");
        }
    }

    message_receiver(nullptr);
}

bool RAWChannelResource::Receive(unsigned char* receive_buffer, uint32_t receive_buffer_capacity,
                                 uint32_t& receive_buffer_size, Locator& remote_locator, Locator& local_locator) {
    try {
        const uint32_t raw_len = receive_buffer_capacity + MAX_IP_HEADER_LENGTH + sizeof(struct udphdr);
        std::vector<unsigned char> buffer(raw_len);

        int bytes = recvfrom(get_socket(), buffer.data(), buffer.size(), 0, NULL, NULL);
        if (bytes < 0) {
            logDebug(RTPS_MSG_OUT, "recvfrom: " << std::strerror(errno) << " (errno: " << errno << ")");
            return false;
        }

        std::size_t byte_count = static_cast<std::size_t>(bytes);
        if (byte_count < (sizeof(struct iphdr) + sizeof(struct udphdr))) {
            logDebug(RTPS_MSG_OUT, "Packet too small.");
            return false;
        }

        struct iphdr* ip_header = (struct iphdr*)buffer.data();
        struct in_addr source_ip;
        source_ip.s_addr = ip_header->saddr;

        char ip_str[INET_ADDRSTRLEN];
        inet_ntop(AF_INET, &source_ip, ip_str, INET_ADDRSTRLEN);

        size_t ip_header_length = ip_header->ihl * 4;
        struct udphdr* udp_header = (struct udphdr*)(buffer.data() + ip_header_length);

        receive_buffer_size = ntohs(udp_header->len) - sizeof(struct udphdr);
        if (receive_buffer_size > 0) {
            memcpy(receive_buffer, buffer.data() + ip_header_length + sizeof(struct udphdr), receive_buffer_size);
            // printf("UDP Data: ");
            // for (uint32_t i = 0; i < receive_buffer_size; i++) {
            //     printf("%02x ", receive_buffer[i]);
            // }
            // This is not necessary anymore but it's left here for back compatibility with versions
            // older than 1.8.1
            if ((receive_buffer_size == 13) && (memcmp(receive_buffer, "EPRORTPSCLOSE", 13) == 0)) {
                return false;
            }

            remote_locator.kind = LOCATOR_KIND_UDPv4;
            (void)IPLocator::setPhysicalPort(remote_locator, ntohs(udp_header->source));
            (void)IPLocator::setIPv4(remote_locator, ip_str);

            local_locator.kind = LOCATOR_KIND_INVALID;
            (void)IPLocator::setPhysicalPort(local_locator, ntohs(udp_header->dest));
        }
        return (receive_buffer_size > 0);
    } catch (const std::exception& error) {
        (void)error;
        logInfo(RTPS_MSG_OUT,
                "Error receiving data: " << error.what() << " - " << message_receiver() << " (" << this << ")");
        return false;
    }
}

void RAWChannelResource::release() {
    // Disable receives on the socket.
    // shutdown always returns a 'shutdown: Transport endpoint is not connected' error,
    int ret = shutdown(get_socket(), SHUT_RDWR);
    if (ret < 0) {
        logDebug(RTPS_MSG_OUT, "shutdown file: " << std::strerror(errno) << " (errno: " << errno << ")");
    }
}

}  // namespace transport
}  // namespace vbs
